package dmp.llapi.util;

import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;


public class FlexApiTask extends Configuration implements Tool {
	
	private Log log = LogFactory.getLog(FlexApiMapper.class);
	
	private String in = "/data/chenjinghui/urlData.txt";
	private String out = "/data/chenjinghui/temp/flexApi";
	private Configuration conf = new Configuration();
	
	@Override
	public void setConf(Configuration conf) {
		this.conf=conf;
	}

	@Override
	public Configuration getConf() {
		return conf;
	}

	@Override
	public int run(String[] args) throws Exception {
		
		//加载配置文件进入内存，用conf传递参数
		List<String> domainList = UrlParseList.elements();
		String[] domains = new String[domainList.size()];
		conf.setStrings("domains", domainList.toArray(domains));
		
		log.info("gggg"+domains.length);
		
		Job job = Job.getInstance(conf);
		job.setJarByClass(FlexApiTask.class);
		job.setMapperClass(FlexApiMapper.class);
		job.setReducerClass(FlexApiReducer.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		Path inPath = new Path(in);
		Path outPath = new Path(out);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);

		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		
		
		return job.waitForCompletion(true) ? 0 : 1;
	}
	
}
